package com.ml.ws.handler;

import java.util.concurrent.ConcurrentHashMap;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.ml.data.PushData;
import com.ml.ws.model.PongMessage;
import com.ml.ws.model.RequestParam;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * @author Mr.D
 * @Date 2020年6月23日
 */

@Slf4j
@Component
public class TradingWebSockeHandler extends SimpleChannelInboundHandler<Object> {

    private WebSocketServerHandshaker handshaker;
    private static final String WEB_SOCKET_URL_TRADE = "ws://127.0.0.1:8111/";
    private static final String TRADE = "/trade";

    @Autowired
    private PushData pushData;
    @Autowired
    private ConcurrentHashMap<Channel, RequestParam> depthMap;
    @Autowired
    private ConcurrentHashMap<Channel, RequestParam> klineMap;
    @Autowired
    private ConcurrentHashMap<Channel, RequestParam> tradeMap;
    @Autowired
    private ConcurrentHashMap<Channel, Integer> topicMap;



    /**
     * 客户端与服务端创建链接的时候调用
     */
    @Override
    public void channelActive(ChannelHandlerContext context) throws Exception {
        log.info("=====================客户端与服务端连接开启=================");
        if(topicMap==null) {
        	System.err.println("111");
        }
        if(context.channel()==null) {
        	System.err.println("222");
        }
        topicMap.put(context.channel(), 0);
        log.info("连接（user）  数量====={}", topicMap.size());
        log.info("=====================end===============================");
    }

    /**
     * 客户端与服务端断开连接的时候调用
     */
    @Override
    public void channelInactive(ChannelHandlerContext context) throws Exception {
        if (klineMap.size() > 0 && klineMap.containsKey(context.channel())) {
            klineMap.remove(context.channel());
        }
        if (tradeMap.size() > 0 && tradeMap.containsKey(context.channel())) {
            tradeMap.remove(context.channel());
        }
        if (depthMap.size() > 0 && depthMap.containsKey(context.channel())) {
            depthMap.remove(context.channel());
        }
        topicMap.remove(context.channel());
        log.info("=====================客户端与服务端连接关闭=================");
        log.info("订阅（kline） 数量====={}", klineMap.size());
        log.info("订阅（deep）  数量====={}", depthMap.size());
        log.info("订阅（market）数量====={}", tradeMap.size());
        log.info("连接（user）  数量====={}", topicMap.size());
        log.info("=====================end===============================");
    }

    /**
     * 服务端接收客户端发送过来的数据结束之后调用
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext context) throws Exception {
        context.flush();
    }

    /**
     * 工程出现异常的时候调用
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext context, Throwable throwable) throws Exception {
        throwable.printStackTrace();
        if (tradeMap.containsKey(context.channel())) {
            tradeMap.remove(context.channel());
        }
        if (depthMap.containsKey(context.channel())) {
            depthMap.remove(context.channel());
        }
        if (klineMap.containsKey(context.channel())) {
            klineMap.remove(context.channel());
        }
        topicMap.remove(context.channel());
        context.close();
    }

    /**
     * 服务端处理客户端websocke请求的核心方法
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {
        // 处理客户端向服务端发起的http握手请求
        if (o instanceof FullHttpRequest) {
            handHttpRequest(channelHandlerContext, (FullHttpRequest) o);
        } else if (o instanceof WebSocketFrame) {// 处理websocket链接业务
            handWebSocketFrame(channelHandlerContext, (WebSocketFrame) o);
        }
    }

    /**
     * 处理客户端与服务端之间的websocket业务
     *
     * @param context
     * @param webSocketFrame
     */
    private void handWebSocketFrame(ChannelHandlerContext context, WebSocketFrame webSocketFrame) {
        if (webSocketFrame instanceof CloseWebSocketFrame) {// 判断是否是关闭websocket的指令
            handshaker.close(context.channel(), (CloseWebSocketFrame) webSocketFrame.retain());
        }
        if (webSocketFrame instanceof PingWebSocketFrame) {// 判断是否是ping消息
            context.channel().write(new PongWebSocketFrame(webSocketFrame.content().retain()));
            return;
        }
        if (webSocketFrame instanceof TextWebSocketFrame) {
            String text = ((TextWebSocketFrame) webSocketFrame).text();
            if (text.contains("cmd") && text.contains("topic")) {
                log.info(text);
                try {
                    RequestParam requestMessage = JSONObject.parseObject(text, RequestParam.class);
                    switch (requestMessage.getCmd()) {
                        case "req":
                            // 一次性请求
                            pushData.handlerReq(context, requestMessage);
                            break;
                        case "sub":
                            pushData.handlerSub(context, requestMessage);
                            break;
                        case "unsub":
                            // 取消订阅
                            pushData.handlerUnsub(context, requestMessage);
                            break;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }else if(text.contains("pong")){
                try {
                    PongMessage pong = JSONObject.parseObject(text, PongMessage.class);
                    // 暗号对上初始化未响应次数
                    if(pong.getPong().equals(HeartBeat.timeMessage)){
                        topicMap.put(context.channel(),0);
                    }
                }catch (Exception e){
                    log.error("错误的消息体:" + text);
                }
            } else {
                log.error("错误请求:" + text);
            }
        }

    }

    /**
     * 处理客户端向服务端发起http握手请求业务
     */
    private void handHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {

        if (!req.decoderResult().isSuccess() || !("websocket".equals(req.headers().get("Upgrade")))) {// 判断是否http握手请求
            sendHttpResponse(ctx, req,
                    new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }
        try {
            if (req.method() != HttpMethod.GET) {
                sendHttpResponse(ctx, req,
                        new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.FORBIDDEN));
                return;
            }
            if (!req.uri().equals(TRADE)) {
                sendHttpResponse(ctx, req,
                        new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.FORBIDDEN));
                return;
            }
            WebSocketServerHandshakerFactory factory = new WebSocketServerHandshakerFactory(WEB_SOCKET_URL_TRADE, null,
                    false);
            handshaker = factory.newHandshaker(req);
            if (handshaker == null) {
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            } else {
                handshaker.handshake(ctx.channel(), req);
            }
        } catch (Exception e) {
            req.release();
        }
    }

    /**
     * 服务端向客户端发送响应消息
     *
     * @param context
     * @param fullHttpRequest
     * @param defaultFullHttpResponse
     */
    private void sendHttpResponse(ChannelHandlerContext context, FullHttpRequest fullHttpRequest,
                                  FullHttpResponse defaultFullHttpResponse) {
        if (defaultFullHttpResponse.status().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(defaultFullHttpResponse.status().toString(), CharsetUtil.UTF_8);
            defaultFullHttpResponse.content().writeBytes(buf);
            buf.release();
        }

        // 服务端向客户端发送数据
        ChannelFuture future = context.channel().writeAndFlush(defaultFullHttpResponse);
        if (defaultFullHttpResponse.status().code() != 200) {
            future.addListener(ChannelFutureListener.CLOSE);
        }

    }


}
